// SPDX-License-Identifier: Mulan PSL v2
/*
 * Copyright (c) 2025 Huawei Technologies Co., Ltd.
 * This software is licensed under Mulan PSL v2.
 * You can use this software according to the terms and conditions of the Mulan PSL v2.
 * You may obtain a copy of Mulan PSL v2 at:
 *         http://license.coscl.org.cn/MulanPSL2
 *
 * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
 * EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
 * MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
 * See the Mulan PSL v2 for more details.
 */

//! A shared-memory (SHMEM) transport layer implementation.
//!
//! This module provides a high-performance, inter-process communication mechanism
//! built on shared memory ring buffers. It uses one channel for each direction of
//! communication. Synchronization is managed by `parking_lot` primitives placed
//! within the shared memory segment itself.

use std::{
    io, thread,
    time::{Duration, Instant},
};

use tracing::debug;

use crate::ipc::transport::{Endpoint, Transport};

mod error;
pub use error::ShmemTransportError;

mod sys;
use sys::shmem::Shmem;

mod channel;

#[cfg(test)]
pub use sys::shmem::unique_shmem_name;

const S2C_SUFFIX: &str = "_s2c";
const C2S_SUFFIX: &str = "_c2s";

#[derive(Debug, Clone, Copy)]
pub struct ShmemTransport {
    buffer_size: usize,
}

impl ShmemTransport {
    #[inline]
    pub fn new(buffer_size: usize) -> Self {
        Self { buffer_size }
    }
}

impl Default for ShmemTransport {
    fn default() -> Self {
        Self::new(2 * sys::page::size())
    }
}

impl Transport for ShmemTransport {
    type Error = ShmemTransportError;
    type Endpoint = ShmemEndpoint;
    type Address = str;

    fn create(&self, addr: &Self::Address) -> Result<Self::Endpoint, Self::Error> {
        let header_size = channel::header_size();
        let data_size = self.buffer_size;

        debug!("[Shmem] Hosting '{}'...", addr);
        let tx_name = format!("{addr}{S2C_SUFFIX}");
        let rx_name = format!("{addr}{C2S_SUFFIX}");

        let (_, tx_writer) = Shmem::create(&tx_name, header_size, data_size)
            .and_then(channel::channel)
            .map_err(|err| ShmemTransportError::ShmemCreateError {
                name: tx_name,
                error: err,
            })?;

        let (rx_reader, _) = Shmem::create(&rx_name, header_size, data_size)
            .and_then(channel::channel)
            .map_err(|err| ShmemTransportError::ShmemCreateError {
                name: rx_name,
                error: err,
            })?;

        let endpoint = ShmemEndpoint {
            reader: rx_reader,
            writer: tx_writer,
        };
        debug!("[Shmem] Endpoint: {:#?}", endpoint);
        debug!("[Shmem] Hosted on '{}'.", addr);

        Ok(endpoint)
    }

    fn connect(
        &self,
        addr: &Self::Address,
        timeout: Duration,
    ) -> Result<Self::Endpoint, Self::Error> {
        const CONNECT_POLL_INTERVAL: Duration = Duration::from_millis(50);
        let header_size = channel::header_size();
        let start = Instant::now();

        debug!("[Shmem] Connecting to '{}'...", addr);

        let tx_name = format!("{addr}{C2S_SUFFIX}");
        let rx_name = format!("{addr}{S2C_SUFFIX}");

        loop {
            if start.elapsed() >= timeout {
                return Err(ShmemTransportError::ConnectTimeout);
            }

            let tx_result = Shmem::open(&tx_name, header_size);
            let rx_result = Shmem::open(&rx_name, header_size);

            match (tx_result, rx_result) {
                (Ok(tx_shmem), Ok(rx_shmem)) => {
                    debug!("[Shmem] Both shmem segments are open, creating channels...");

                    let (_, tx_writer) = channel::channel(tx_shmem).map_err(|e| {
                        ShmemTransportError::ShmemOpenError {
                            name: tx_name,
                            error: e,
                        }
                    })?;

                    let (rx_reader, _) = channel::channel(rx_shmem).map_err(|e| {
                        ShmemTransportError::ShmemOpenError {
                            name: rx_name,
                            error: e,
                        }
                    })?;

                    let endpoint = ShmemEndpoint {
                        reader: rx_reader,
                        writer: tx_writer,
                    };
                    debug!("[Shmem] Endpoint created: {:#?}", endpoint);
                    debug!("[Shmem] Connected to '{}'.", addr);

                    return Ok(endpoint);
                }

                (Err(e), _) if e.kind() != io::ErrorKind::NotFound => {
                    return Err(ShmemTransportError::ShmemOpenError {
                        name: tx_name,
                        error: e,
                    });
                }
                (_, Err(e)) if e.kind() != io::ErrorKind::NotFound => {
                    return Err(ShmemTransportError::ShmemOpenError {
                        name: rx_name,
                        error: e,
                    });
                }
                _ => {
                    thread::sleep(CONNECT_POLL_INTERVAL);
                }
            }
        }
    }
}

/// A `Transport` implementation that uses two shared memory channels for bidirectional communication.
#[derive(Debug)]
pub struct ShmemEndpoint {
    reader: channel::ShmemReader,
    writer: channel::ShmemWriter,
}

impl Endpoint for ShmemEndpoint {
    type Error = ShmemTransportError;
    type ReadBuf<'a> = channel::ShmemReadGuard<'a>;
    type WriteBuf<'a> = channel::ShmemWriteGuard<'a>;

    fn read_timeout(&self, timeout: Duration) -> Result<Self::ReadBuf<'_>, Self::Error> {
        self.reader.read_timeout(timeout)
    }

    fn read(&self) -> Result<Self::ReadBuf<'_>, Self::Error> {
        self.reader.read()
    }

    fn write_timeout(&self, timeout: Duration) -> Result<Self::WriteBuf<'_>, Self::Error> {
        self.writer.write_timeout(timeout)
    }

    fn write(&self) -> Result<Self::WriteBuf<'_>, Self::Error> {
        self.writer.write()
    }
}
